package com.lagou.client;

import com.lagou.handler.UserClientHandler;
import com.lagou.service.pojo.RpcRequest;
import com.lagou.serialize.JSONSerializer;
import com.lagou.serialize.RpcDecoder;
import com.lagou.serialize.RpcEncoder;
import com.lagou.service.util.ZookeeperUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 消费者
 */
public class RPCConsumer {


    /**
     * 用于保存ipPort信息和bootstrap关系
     */
    private static Map<String, Bootstrap> map = new HashMap<>();

    /**
     * 初始化的注册中心列表
     */
    private static List<String> pathString;

    //中间变量
    private static List<String> initIpPortList;

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    //2.声明一个自定义事件处理器  UserClientHandler
    private static UserClientHandler userClientHandler;


    //3.编写方法,初始化客户端  ( 创建连接池  bootStrap  设置bootstrap  连接服务器)
    public static void initClient() throws Exception {
        //1) 初始化UserClientHandler
        userClientHandler = new UserClientHandler();
        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)从zookeeper中获取服务端ipPort信息
        String level1Path = "/registerCenter";
        //1.与zookeeper创建连接
        CuratorFramework client = ZookeeperUtil.createSession("192.168.174.128:2181");
        //2.获取zookeeper注册列表节点/registerCenter下的服务端ipPort信息
        pathString = getServerIpPortFromZookeeper(client, level1Path);
        //赋值
        initIpPortList = pathString;
        //3.与服务端建立连接
        if (pathString != null && pathString.size() > 0) {
            //获取服务端的个数
            int size = pathString.size();
            List<Bootstrap> bootstrapList = new ArrayList<>(size);
            //创建并初始化bootstrap
            pathString.forEach(path -> {
                Bootstrap bootstrap = new Bootstrap();
                configurateBootStrap(bootstrap, group);
                bootstrapList.add(bootstrap);
            });
            for (int i = 0; i < pathString.size(); i++) {
                String path = pathString.get(i);
                String[] split = path.split(":");
                String ip = split[0];
                int port = Integer.parseInt(split[1]);
                Bootstrap bootstrap = bootstrapList.get(i);
                bootstrap.connect(new InetSocketAddress(ip, port));
                map.put(path, bootstrap);
                System.out.println("bootstrap【" + bootstrap.hashCode() + "】连接成功，端口号是：" + path);
            }
            //5.todo 并在该节点下注册监听
            registerListener(client, level1Path);
        } else {
            System.out.println("连接服务端失败");
        }
    }

    private static void configurateBootStrap(Bootstrap bootstrap, EventLoopGroup group) {
        bootstrap.group(group)
                //设置通道为NIO
                .channel(NioSocketChannel.class)
                //设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY, true)
                //监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //设置json编码器
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new StringDecoder());
                        //添加自定义事件处理器
                        pipeline.addLast(userClientHandler);
                    }
                });
    }

    /**
     * 注册监听
     */
    private static void registerListener(CuratorFramework client, String level1Path) throws Exception {
        //创建PathChildrenCache
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, level1Path, true);
        //BUILD_INITIAL_CACHE 代表使用同步的方式进行缓存初始化。
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                PathChildrenCacheEvent.Type eventType = event.getType();
                //获取注册列表
                List<String> registerPathList = getServerIpPortFromZookeeper(client, level1Path);
                System.out.println("当前注册中心节点为：" + registerPathList);
                switch (eventType) {
                    case CONNECTION_RECONNECTED:
                        pathChildrenCache.rebuild();
                        break;
                    case CONNECTION_SUSPENDED:
                        break;
                    case CONNECTION_LOST:
                        System.out.println("Connection lost");
                        break;
                    case CHILD_ADDED:
                        System.out.println("监听到有子节点添加");
                        //获取添加的服务端信息
                        registerPathList.removeAll(initIpPortList);
                        String addIpPort = registerPathList.get(0);
                        String[] split = addIpPort.split(":");
                        String ip = split[0];
                        int port = Integer.parseInt(split[1]);
                        //4.再次连接服务端
                        Bootstrap addBootstrap = map.get(addIpPort);
                        addBootstrap.connect(ip, port);
                        System.out.println("bootstrap【" + addBootstrap.hashCode() + "】连接成功，端口号是：" + addIpPort);
                        //重置初始化的注册列表
                        initIpPortList = pathString;
                        break;
                    case CHILD_REMOVED:
                        System.out.println("监听到有子节点移除");
                        //获取移除的服务端信息
                        initIpPortList.removeAll(registerPathList);
                        String removeIpPort = initIpPortList.get(0);
                        //获取当前注册列表
                        initIpPortList = pathString;
                        initIpPortList.removeIf(i -> Objects.equals(i, removeIpPort));
                        Bootstrap removeBootstrap = map.get(removeIpPort);
                        System.out.println("bootstrap【" + removeBootstrap.hashCode() + "】下线，端口号是：" + removeIpPort);
                    case CHILD_UPDATED:
                        System.out.println("Child updated");
                        break;
                    default:
                        break;
                }
            }
        });

    }

    /**
     * 获取zookeeper注册列表节点/registerCenter下的服务端ipPort信息
     */
    private static List<String> getServerIpPortFromZookeeper(CuratorFramework client, String level1Path) throws Exception {
        return client.getChildren().forPath(level1Path);
    }


    //4.编写一个方法,使用JDK的动态代理创建对象
    // serviceClass 接口类型,根据哪个接口生成子类代理对象;   providerParam :  "UserService#sayHello#"
    public static Object createProxy(Class<?> serviceClass, final RpcRequest rpcRequest) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object o, Method method, Object[] objects) throws Throwable {
                        //1)初始化客户端cliet
                        if (userClientHandler == null) {
                            initClient();
                        }

                        //2)给UserClientHandler 设置param参数
                        userClientHandler.setRpcRequest(rpcRequest);

                        //3).使用线程池,开启一个线程处理处理call() 写操作,并返回结果
                        Object result = executorService.submit(userClientHandler).get();

                        //4)return 结果
                        return result;
                    }
                });
    }

}
